package com.avicit.zipkinkafkaesserver.dao.es;

import com.avicit.zipkinkafkaesserver.dao.es.entity.ZipkinInterfaceCountES;
import com.avicit.zipkinkafkaesserver.dto.ZipkinInterfaceCountDto;
import com.avicit.zipkinkafkaesserver.entity.ZipkinInterfaceCount;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.filter.InternalFilter;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.metrics.sum.InternalSum;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.*;

/**
 * es dao
 */
@Slf4j
@Repository
public class ZipkinInterfaceCountEsDao extends BaseEsDao<ZipkinInterfaceCountES> {

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Override
    ElasticsearchTemplate elasticsearchTemplate() {
        return elasticsearchTemplate;
    }

    public List<ZipkinInterfaceCount> searchFromEs(long startTime, long endTime, String index, String type) {
        //根据createTime 查询
//        RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("createTime");
//        rangeQueryBuilder.from(startTime).to(endTime)
//                .includeLower(true)//允许取边界值
//                .includeUpper(true);
//        SearchResponse response = elasticsearchTemplate.getClient().prepareSearch(index)
//                .setTypes(type)
//                .setQuery(rangeQueryBuilder)
//                .addAggregation(AggregationBuilders.filter("filter",rangeQueryBuilder)//设置过滤条件
//                        .subAggregation(AggregationBuilders.terms("queryKeyGroupBy").field("queryKey"))//根据 queryKey 进行group by
//                        .subAggregation(AggregationBuilders.sum("sum").field("interfaceCount")))//求聚合
//                .execute().actionGet();
//
//        SearchHits hits = response.getHits();
//        Aggregations aggregations = response.getAggregations();
//        Aggregation filter = aggregations.get("filter");


//        List<ZipkinInterfaceCount> result = new ArrayList<>();
//        hits.forEach(item->{
//            Map<String, Object> sourceAsMap = item.getSourceAsMap();
//            result.add(ZipkinInterfaceCount
//                    .builder()
//                    .interfaceIp((String) sourceAsMap.get("interfaceIp"))
//                    .serverName((String) sourceAsMap.get("serverName"))
////                    .serverPort((Integer))
//                    .interfaceStatus((String) sourceAsMap.get("interfaceStatus"))
//                    .interfacePath((String) sourceAsMap.get("interfacePath"))
//                    .createTime((Long)sourceAsMap.get("createTime"))
//                    .id(item.getId())
//                    .build());
//        });
//        return result;
        return null;
    }

    /**
     * 查询 接口的 调用次数
     *
     * @param index
     * @param type
     * @return
     */
    public Map<String, Double> doInterfaceCount(long startTime, long endTime, String index, String type) {
        //根据createTime 查询
        startTime = startTime==0?(System.currentTimeMillis()-5*24*60*60*1000):startTime;
        RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("createTime");
        rangeQueryBuilder.from(startTime*1000).to(endTime == 0?System.currentTimeMillis()*1000: endTime*1000)
                .includeLower(true)//允许取边界值
                .includeUpper(true);
        SearchResponse response = elasticsearchTemplate.getClient().prepareSearch(index)
                .setTypes(type)
                .setQuery(rangeQueryBuilder)
                .addAggregation(AggregationBuilders.filter("filter", rangeQueryBuilder)//设置过滤条件
                        .subAggregation(AggregationBuilders.terms("queryKeyGroupBy").field("queryKey.keyword").subAggregation(AggregationBuilders.sum("sum").field("interfaceCount"))))// xx.keyword 设置该字段不分词
                .execute().actionGet();//根据 queryKey 进行group by

        Map<String, Double> resultMap = new HashMap<>();
        Map<String, Aggregation> stringAggregationMap = response.getAggregations().asMap();
        InternalFilter internalFilter = (InternalFilter) stringAggregationMap.get("filter");
        List<Aggregation> aggregations = internalFilter.getAggregations().asList();
        aggregations.forEach(item -> {
            StringTerms stringTerms = (StringTerms) item;
            List<StringTerms.Bucket> buckets = stringTerms.getBuckets();
            buckets.forEach(bucket -> {
                String keyAsString = bucket.getKeyAsString();
                InternalSum sum = bucket.getAggregations().get("sum");//统计结果
                double value = sum.getValue();
                resultMap.put(keyAsString, value);
            });
        });
        return resultMap;
    }


    /**
     * 根据 esEndTime 的最大值获取数据,根据 esEndTime 倒排，取1条
     * @return
     */
    public ZipkinInterfaceCountES findByBatchNumber(String index, String type) throws IOException {
        XContentBuilder mapping = this.getxContentBuilderMapping();
        boolean create = ElasticsearchTemplateWarpper.builder(elasticsearchTemplate).createIndex(index, type, mapping);
        SearchRequestBuilder searchRequestBuilder = elasticsearchTemplate.getClient().prepareSearch(index);
        //根据 esEndTime 的最大值获取数据,根据 esEndTime 倒排，取1条
        FieldSortBuilder esEndTime = SortBuilders.fieldSort("esEndTime").order(SortOrder.DESC);
        SearchResponse response = searchRequestBuilder
                .setTypes(type)
                .addSort(esEndTime)
                .setSize(1)
                .execute().actionGet();
        ZipkinInterfaceCountES build = ZipkinInterfaceCountES.builder().build();
        response.getHits().forEach(item->{
            Map<String, Object> sourceAsMap = item.getSourceAsMap();
                build.setId(item.getId());
                build.setQueryKey((String) sourceAsMap.get("queryKey"));
                build.setInterfaceCount((Integer) sourceAsMap.get("interfaceCount"));
                build.setServerName((String) sourceAsMap.get("serverName"));
                build.setInterfaceIp((String) sourceAsMap.get("interfaceIp"));
                build.setInterfacePath((String) sourceAsMap.get("interfacePath"));
                build.setEsEndTime((Long) sourceAsMap.get("esEndTime"));
                build.setEsStartTime((Long) sourceAsMap.get("esStartTime"));
        });
        return build;
    }

    /**
     * 构建 mapping
     * 来自官方代码案例
     * @return
     * @throws IOException
     */
    private XContentBuilder getxContentBuilderMapping() throws IOException {
        //设置mapping,防止在新创建的index 时，还没添加数据导致的查询异常
        XContentBuilder builder = XContentFactory.jsonBuilder();
        builder.startObject();
        {
            builder.startObject("properties");
            {
                builder.startObject("batchNumber");
                {
                    builder.field("type", "long");
                }
                builder.endObject();
                builder.startObject("esEndTime");
                {
                    builder.field("type", "long");
                }
                builder.endObject();
                builder.startObject("esStartTime");
                {
                    builder.field("type", "long");
                }
                builder.endObject();
            }
            builder.endObject();
        }
        builder.endObject();
        return builder;
    }

//    /**
//     * 更新
//     * @param updateMap
//     * @param index
//     * @param type
//     * @param id es 自己的id
//     */
//    public void update(Map<String,Object> updateMap, String index, String type,String id) {
//        UpdateQueryBuilder updateQueryBuilder = new UpdateQueryBuilder();
//        UpdateQuery updateQuery = updateQueryBuilder.withIndexName(index)
//                .withType(type)
//                .withId(id)
//                .build();
//        UpdateRequest updateRequest = new UpdateRequest();
//        updateRequest.doc(updateMap);
//        updateQuery.setUpdateRequest(updateRequest);
//        UpdateResponse update = elasticsearchTemplate.update(updateQuery);
//    }

    /**
     * 统计接口一天的调用总次数
     * @return
     */
    public List<ZipkinInterfaceCountDto> getCountResultFromEs(String index, String type) {
        ElasticsearchTemplateWarpper.builder(elasticsearchTemplate).createIndex(index);
        //求和
        SearchResponse response = elasticsearchTemplate.getClient().prepareSearch(index)

                .addAggregation(AggregationBuilders.terms("interfaceKey").field("interfaceKey.keyword")//分组
                        .subAggregation(AggregationBuilders.sum("sumInterfaceKey").field("interfaceCount"))
                        .subAggregation(AggregationBuilders.terms("status").field("interfaceStatus.keyword")
                                .subAggregation(AggregationBuilders.sum("errorOk").field("interfaceCount"))
                        ))//状态分组
                .setTypes(type).execute().actionGet();
        List<Aggregation> aggregations = response.getAggregations().asList();
        List<ZipkinInterfaceCountDto> zipkinInterfaceCountDtos = new ArrayList<>();
        aggregations.forEach(item->{
            if (item instanceof StringTerms) {
            StringTerms stringTerms = (StringTerms) item;
            List<StringTerms.Bucket> buckets = stringTerms.getBuckets();

            buckets.forEach(bucket -> {
                ZipkinInterfaceCountDto interfaceCountDto = ZipkinInterfaceCountDto.builder().build();
                String interfaceMsg =(String) bucket.getKey();
                interfaceCountDto.setInterfaceKey(interfaceMsg);
                //error/ok 统计信息
                Map<String, Integer> okErrorMap = new HashMap<>();

                bucket.getAggregations().forEach(aggregation -> {
                    if (aggregation instanceof InternalSum) {
                        InternalSum internalSum = (InternalSum) aggregation;
                        int allCount = (int) internalSum.getValue();//服务调用总数
                        interfaceCountDto.setInterfaceCount(allCount);
                    }
                    if (aggregation instanceof StringTerms) {
                        StringTerms terms = (StringTerms) aggregation;
                        terms.getBuckets().forEach(okError->{
                            InternalSum internalSum = okError.getAggregations().get("errorOk");
                            okErrorMap.put(okError.getKeyAsString(),(int)internalSum.getValue());
                        });
                    }
                });
                interfaceCountDto.setErrorOk(okErrorMap);
                log.info("接口统计信息 :{}", interfaceCountDto);
                zipkinInterfaceCountDtos.add(interfaceCountDto);
            });
        }});
        return zipkinInterfaceCountDtos;
    }
}
